/*
 * dtalk_engine.h
 *
 *  Created on: 2021年10月14日
 *      Author: guyadong
 */

#ifndef DTALK_DTALK_ENGINE_H_
#define DTALK_DTALK_ENGINE_H_
#include "dtalk_item.h"
#include "dtalk_common_utils.h"
#include <simplemq/simplemq.hpp>
#include <simplemq/simplemq_commons.hpp>
#include <cpptime/cpptime.h>
#include <md5/md5.hpp>

#define DTALK_CPP_VERSION_MAJOR 0
#define DTALK_CPP_VERSION_MINOR 0
#define DTALK_CPP_VERSION_PATCH 0
#define DTALK_CPP_VERSION_SNAPSHOT 1
#define DTALK_CPP_VERSION ((DTALK_CPP_VERSION_MAJOR<<16)|(DTALK_CPP_VERSION_MINOR << 8) |DTALK_CPP_VERSION_PATCH)

namespace dtalk { namespace engine {
	class BaseItemEngine;
	using item_engine_ptr = std::shared_ptr<BaseItemEngine>;
	struct BaseItemDriver;
	using item_driver_ptr = std::shared_ptr<BaseItemDriver>;

	/**
	 * 菜单驱动接口
	 */
	struct BaseItemDriver : public virtual simplemq::IMessageAdapter{
		/**
		 * @return 返回根菜单实例
		 */
		virtual item_ptr_type getRoot() const = 0;
		/**
		 * 设置当前设备的MAC地址(HEX字符串)
		 * @param selfMac 要设置的 selfMac
		 * @return 当前接口对象
		 */
		virtual void setSelfMac(const std::string &selfMac) = 0;
		/**
		 * @return 返回最近一次收到消息的时间戳
		 */
		virtual date::time_mills lastHitTime() const = 0;

		/**
		 * 指定请求响应频道
		 * @param ackChannelName 频道名
		 */
		virtual void setAckChannel(const std::string &ackChannelName) {};
		virtual ~BaseItemDriver() = default;
	};
	/**
	 * 消息驱动的菜单引擎，根据收到的请求执行对应的动作<br>
	 * <ul>
	 * <li>OPTION：修改指定的参数</li>
	 * <li>CMD:执行指定的命令</li>
	 * <li>MENU:输出菜单内容</li>
	 * </ul>
	 *
	 */
	class BaseItemEngine : public virtual BaseItemDriver,public std::enable_shared_from_this<BaseItemEngine>
	{
	private:
		item_ptr_type currentLevel;
		item_ptr_type root;
		/**
		 * 当前设备的MAC地址(HEX字符串)
		 */
		std::string selfMac;
		/**
		 * 最近一次操作的时间戳
		 */
		date::time_mills lasthit;
		/**
		 * 当前执行的设备命令
		 */
		std::string cmdLock;
		std::mutex checker_mtx;
		CppTime::timer_id task_id;
		CppTime::Timer timer;
		cmd_status_listener_ptr dtalkListener;
		friend struct DtalkListener;
	protected:
		/**
		 * 发送菜单数据
		 * @param object 菜单实例
		 */
		virtual void responseMenu(const item_ptr_type& object) = 0;
		/**
		 * 发送ack消息
		 * @param ack 响应实例
		 */
		virtual void responseAck(const Ack &ack) = 0;
		virtual void setAckChannel(const std::string &ackChannelName) override {};
		/**
		 * 用于ICmdInteractiveStatusListener发送CMD命令的ack消息
		 * @param ack 响应实例
		 */
		virtual void responseCmdAck(const Ack &ack) = 0;
		virtual void beforeSubscribe(const dtalk::json&j) 
		{
			auto reqType = json_utilits::get_value(j, REQ_FIELD_REQTYPE, ReqCmdType::DRQ_DEFAULT);
			setReqType(reqType);
		}
		virtual void afterSubscribe() { setReqType(ReqCmdType::DRQ_DEFAULT); }
		void startInternalCmdChecker() {
			std::lock_guard<std::mutex> guard(checker_mtx);
			if (task_id != -1)
			{
				timer.remove(task_id);
			}
			auto listener = std::dynamic_pointer_cast<ICmdInteractiveStatusListener>(getDtalkListener());
			auto period = std::chrono::seconds(listener->getProgressInternal());
			task_id = timer.add(period, [period, listener,this](CppTime::timer_id) {
				/** cmdLock 解锁状态下自动中断定时任务 */				
				if (!cmdLock.empty()) {
					auto lastInternel = std::chrono::system_clock::now() - listener->getLastProgress();
					if (lastInternel > period * 4) {
						listener->onTimeout();
					}
				}
			}, period);
		}
		cmd_status_listener_ptr getDtalkListener();
		ReqCmdType& getReqType()
		{
			static thread_local ReqCmdType reqType = ReqCmdType::DRQ_DEFAULT;
			return reqType;
		}
		void setReqType(ReqCmdType reqType) {
			getReqType() = reqType;
		}
	public:
		BaseItemEngine() : root(make_item_shared<RootMenu>()), task_id(-1) {}
		item_ptr_type getRoot() const override 
		{
			return root;
		}
		item_engine_ptr setRoot(const item_ptr_type &root)
		{
			throw_if_null(root);
			this->root = root;
			/** 自动添加退出命令在最后 */
			if (nullptr == this->root->getChild(QUIT_NAME))
			{
				auto quit = makeQuit();
				this->root->addChild(quit);
			}
			return shared_from_this();
		}
		std::string getSelfMac()const 
		{
			return selfMac;
		}
		void setSelfMac(const std::string &selfMac) override
		{
			this->selfMac = selfMac;
		}
		date::time_mills lastHitTime() const override 
		{
			return lasthit;
		}
		void setLastHitTime(const date::time_mills &lastHit) 
		{
			this->lasthit = lastHit;
		}

		virtual void onSubscribe(const dtalk::json &input) override
		{
			bool isQuit = false;
			Ack ack;
			dtalk::json j = input;
			try {
				auto now = std::chrono::system_clock::now();
				lasthit = std::chrono::time_point_cast<std::chrono::milliseconds>(now);
				beforeSubscribe(j);
				auto multiTarget = ReqCmdType::DRQ_MULTI == getReqType();
				auto taskQueue = ReqCmdType::DRQ_TASKQUEUE == getReqType();
				auto parameters = json_utilits::get_value(j, REQ_FIELD_PARAMETERS, dtalk::json());
				/** 删除参数字段以避免解析为item对象时抛出异常 */
				j.erase(REQ_FIELD_PARAMETERS);
				ack.status = Status::ACK_OK;
				ack.deviceMac = selfMac;
				ack.cmdSn = json_utilits::get_value(j, REQ_FIELD_CMDSN, -1);
				normalize(j, currentLevel ? currentLevel : root);
				auto req = dtalk::from_json(j);
				item_ptr_type found;
				if (currentLevel) {
					found = currentLevel->getChild(req->getName());
					if (nullptr == found) {
						found = currentLevel->getChild(req->getPath());
					}
				}
				if (root->getPath() == req->getPath()) {
					found = root;
				}
				if (nullptr == found ) {
					found = root->getChildByPath(req->getPath());
				}
				throw_if_msg(!found, "UNSUPPORTED ITEM");
				throw_if_msg(found->isDisable(), gdface::string_utils::format("DISABLE ITEM [%s]", found->getPath().c_str()));
				/** 交互设备命令没执行完，则抛出异常 */
				throw_except_if_msg(std::logic_error, !cmdLock.empty() && cmdLock != found->getPath(),
					gdface::string_utils::format("CMD LOCKED %s", cmdLock.c_str()));
				ack.item = found->getPath();
				switch (found->getCatalog()) {
				case DIT_OPTION: {
					throw_except_if_msg(std::logic_error, taskQueue, "OPTION item unsupport task request");
					std::static_pointer_cast<BaseOption>(found)->updateFrom(req->as<BaseOption>());
					break;
				}
				case DIT_CMD: {
					if (dtalk::isBack(found)) {
						throw_except_if_msg(std::logic_error,multiTarget, "'back' cmd unsupport multi-target cmd request");
						throw_except_if_msg(std::logic_error, taskQueue, "'back' cmd unsupport task request");

						//  输出上一级菜单
						// currentLevel = found->parent ? found->parent : root; /** 这是Java 版本的逻辑 */
						currentLevel = currentLevel->parent ? currentLevel->parent : root; /** Java 版的逻辑有问题,所以这里没有参照 Java 版本逻辑 */
						responseMenu(currentLevel);
						return;
					}
					else if (dtalk::isQuit(found)) {
						throw_except_if_msg(std::logic_error, multiTarget, "'quit' cmd unsupport multi-target cmd request");
						throw_except_if_msg(std::logic_error, taskQueue, "'quit' cmd unsupport task request");

						isQuit = true;
					}
					else 
					{
						/** 执行命令 */
						auto cmd = std::static_pointer_cast<CmdItem>(found);
						auto reqCmd = std::static_pointer_cast<CmdItem>(req);
						if (cmdLock.empty())
						{
							for (auto param : cmd->getParameters())
							{
								param->updateFrom(reqCmd->getParameter(param->getName()));
							}
							if (cmd->isInteractiveCmd())
							{
								throw_except_if_msg(std::logic_error, multiTarget, "interactive cmd unsupport multi-target cmd request");
								throw_except_if_msg(std::logic_error, taskQueue, "interactive cmd  unsupport task request");
								/** 启动设备交互命令执行 */
								cmd->startInteractiveCmd(getDtalkListener()->init(ack));
								/** 设置为正常启动状态 */ 
								ack.status = Status::ACK_ACCEPTED;
								/** 命令加锁 */ 
								cmdLock = cmd->getPath();
								/** 启动超时检查，避免设备死机造成的锁死 */ 
								startInternalCmdChecker();
							}
							else
							{
								/** 启动设备命令执行 */
								ack.value = (cmd->runImmediateCmd(parameters));
							}
						}
						else 
						{
							/* 只有交互设备命令会加锁，加锁状态下只能执行取消命令 */
							throw_except_if_msg(std::logic_error, !cmd->isInteractiveCmd(), gdface::string_utils::format("NOT INTERACTIVE CMD %s", cmd->getPath().c_str()));
							/** 上一个命令没执行完，则抛出异常 */
							auto c = reqCmd->getField<bool>(_jf_canceled);
							throw_except_if_msg(std::logic_error,  (c && *c), gdface::string_utils::format("CMD REENTRANT  %s", cmdLock.c_str()));
							cmd->cancelInteractiveCmd();
							/** 命令解锁 */ 
							cmdLock = "";
						}
					}
					break;
				}
				case DIT_MENU: {
					throw_except_if_msg(std::logic_error, multiTarget, "MENU item unsupport multi-target cmd request");
					throw_except_if_msg(std::logic_error, taskQueue, "MENU item unsupport task request");
					/** 输出当前菜单后直接返回 */  
					currentLevel = found;
					responseMenu(currentLevel);
					break;
				}
				default:
					throw new std::invalid_argument(gdface::string_utils::format("UNSUPPORTED CATALOG [%s] of ITEM [%s]", name_of(found->getCatalog()).c_str(), found->getPath().c_str()));
				}
			}
			catch(const std::exception & e){
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(),e.what());
				ack.writeError(e);
			}
			try {
				// 向ack频道发送返回值消息
				responseAck(ack);
			}
			catch (const std::exception &e) {
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
			}
			afterSubscribe();
			if (isQuit)
			{
				/** 取消频道订阅,中断连接 */
				throw simplemq::SmqUnsubscribeException(true);
			}
		}
	};
	/**
	 * 消息驱动的菜单引擎实现<br>
	 *
	 */
	struct ItemEngineMqImpl : public virtual BaseItemEngine
	{
	private:
		simplemq::publisher_ptr publisher;
		simplemq::Channel ackChannel;
		simplemq::Channel menuChannel;

		std::string& getAckChannelName()
		{
			static thread_local std::string ackChannelName;
			return ackChannelName;
		}
		void setAckChannelName(const std::string & ackChannelName)
		{
			getAckChannelName() = ackChannelName;
		}
	public:
		ItemEngineMqImpl(const simplemq::publisher_ptr &publisher) : publisher(publisher) {
			throw_if_null(publisher);
		}
		virtual ~ItemEngineMqImpl() = default;
		virtual void setAckChannel(const std::string &name) override
		{
			ackChannel = simplemq::Channel(name);
			menuChannel = simplemq::Channel(name);
		}
		std::string getAckChannel() const {
			return ackChannel.name;
		}
	protected:
		virtual void responseMenu(const item_ptr_type& object) override
		{
			if (object) 
			{
				publisher->publish(menuChannel.name, *object);
			}
		}
		/**
		 * 向ack频道发送返回值消息
		 */
		void responseAck(const Ack& ack) override {
			switch (getReqType()) {
			case DRQ_MULTI:
			case DRQ_TASKQUEUE:
			{
				auto acn = getAckChannelName();
				if (!acn.empty()) {
					publisher->publish(acn, ack);
				}
				break;
			}
			default:
				publisher->publish(ackChannel.name, ack);
				break;
			}
		}
		virtual void responseCmdAck(const Ack &ack) override
		{
			publisher->publish(ackChannel.name, ack);
		}
		void beforeSubscribe(const dtalk::json&j) override {
			BaseItemEngine::beforeSubscribe(j);
			setAckChannelName(json_utilits::get_value(j, REQ_FIELD_ACKCHANNEL, std::string("")));
		}
		void afterSubscribe() override {
			BaseItemEngine::afterSubscribe();
			setAckChannelName("");
		}
	};
	struct DtalkListener : public virtual ICmdInteractiveStatusListener, public std::enable_shared_from_this<DtalkListener>
	{
		Ack ack;
		date::time_mills lastProgress;
		item_engine_ptr engine;
		DtalkListener(const item_engine_ptr &engine) : engine(engine) {
			throw_if_null(engine);
		}
		virtual void onProgress(int progress, const std::string &statusMessage) override
		{
			try
			{
				auto now = std::chrono::system_clock::now();
				lastProgress = std::chrono::time_point_cast<std::chrono::milliseconds>(now);
				engine->responseCmdAck(ack);
			}
			catch (const std::exception& e)
			{
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
			}
			catch (...)
			{
				SAMPLE_ERR("UNKNOW ERROR");
			}
		}
		virtual void onFinished(const dtalk::json &value) override
		{
			try
			{
				/** 命令解锁 */
				engine->cmdLock = "";
				ack.status = Status::ACK_OK;
				ack.value = value;
				engine->responseCmdAck(ack);
			}
			catch (const std::exception& e)
			{
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
			}
			catch (...)
			{
				SAMPLE_ERR("UNKNOW ERROR");
			}
		}
		virtual void onCaneled() override
		{
			try
			{
				/** 命令解锁 */
				engine->cmdLock = "";
				ack.status = Status::ACK_CANCELED;
				engine->responseCmdAck(ack);
			}
			catch (const std::exception& e)
			{
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
			}
			catch (...)
			{
				SAMPLE_ERR("UNKNOW ERROR");
			}
		}
		virtual void onError(std::string errorMessage, const std::exception &error)
		{
			try
			{
				/** 命令解锁 */ 
				engine->cmdLock = "";
				ack.status = dtalk::Status::ACK_ERROR;
				std::string msg;
				if (!errorMessage.empty())
				{
					msg = msg.append(errorMessage);
				}
				const char* w = error.what();
				if (w && std::strlen(w)) {
					msg = msg.append(":").append(w);
				}
				if (!msg.empty())
				{
					ack.statusMessage = msg;
				}
				engine->responseCmdAck(ack);
			}
			catch (const std::exception& e)
			{
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
			}
			catch (...)
			{
				SAMPLE_ERR("UNKNOW ERROR");
			}
		}

		virtual void onTimeout() override {
			try {
				/** 命令解锁 */
				engine->cmdLock = "";
				ack.status = Status::ACK_TIMEOUT;
				engine->responseCmdAck(ack);
			}
			catch (const std::exception& e)
			{
				SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
			}
			catch (...)
			{
				SAMPLE_ERR("UNKNOW ERROR");
			}
		}

		virtual int getProgressInternal() override { return 2; }

		virtual date::time_mills getLastProgress()override { return lastProgress; }

		virtual cmd_status_listener_ptr init(const Ack& ack) override {
			this->ack.item = ack.item;
			this->ack.deviceMac = ack.deviceMac;
			auto now = std::chrono::system_clock::now();
			this->lastProgress = std::chrono::time_point_cast<std::chrono::milliseconds>(now);
			return std::static_pointer_cast<ICmdInteractiveStatusListener>(shared_from_this());
		}
	};

	inline cmd_status_listener_ptr BaseItemEngine::getDtalkListener()
	{
		if (!dtalkListener)
		{
			dtalkListener = std::make_shared<DtalkListener>(shared_from_this());
		}
		return dtalkListener;
	}
	/**
	 * 验证连接请求合法性接口
	 */
	struct RequestValidator
	{
		/**
		* 验证连接请求合法性
		* @param connj 连接请求JSON,包含字段:mac--请求连接设备的物理地址,pwd--连接密码的MD5校验码
		* @param clientID [out] 用于返回发送连接请求CLIENT的MAC地址(HEX格式)或其他可以唯一识别CLIENT的字符串,返回{@code null}或空视为验证失败
		* @throws Exception 连接请求验证不通过
		*/
		virtual void validate(const simplemq::json &connj, std::string &clientID) = 0;
	};
	using req_validator_ptr = std::shared_ptr<RequestValidator>;

	/** 基于密码验证的连接请求 */
	struct ConnectReq 
	{
		/** 请求连接设备的物理地址 */
		std::string mac;
		/** 连接密码的MD5校验码	*/
		std::string pwd;

		ConnectReq(const std::string &mac = "", const std::string &pwd = "") : mac(mac),pwd(pwd) {}
		_BASIC_JSON_TPL_DECLARATION_
		ConnectReq(const _BASIC_JSON_TPL_ & connj) {
			throw_if_msg(connj.is_null(), "NULL REQUEST");
			throw_if_msg(!connj.contains("mac") || !connj["mac"].is_string(), "INVALID REQUEST,'mac' field required");
			throw_if_msg(!connj.contains("pwd") || !connj["pwd"].is_string(), "INVALID REQUEST,'pwd' field required");
			json_utilits::fill_field(mac, connj, "mac");
			json_utilits::fill_field(pwd, connj, "pwd");
		}
		dtalk::json to_json()const
		{
			dtalk::json j;
			j["mac"] = mac;
			j["pwd"] = pwd;
			return j;
		}
		_BASIC_JSON_TPL_DECLARATION_
		void from_json(const _BASIC_JSON_TPL_ & j)
		{
			json_utilits::fill_field(mac, j, "mac");
			json_utilits::fill_field(pwd, j, "pwd");
		}
		std::string toString() { to_json().dump(); }
	};

	_BASIC_JSON_TPL_DECLARATION_
	void to_json(_BASIC_JSON_TPL_ & j, const ConnectReq & connReq)
	{
		j = connReq.to_json();
	}
	_BASIC_JSON_TPL_DECLARATION_
	void from_json(const _BASIC_JSON_TPL_ & j, ConnectReq & connReq)
	{
		connReq.from_json(j);
	}
	class SampleConnector;
	using sample_onnector_ptr = std::shared_ptr<SampleConnector>;
	/**
	 * 设备端连接控制器简单实现<br>
	 * 接收连接请求并验证请求合法性，如果连接请求有效，则将请求频道名封装到{@link Ack}中发送给管理端<br>
	 * 管理端有了请求频道名才可以向设备端发送菜单命令(item)请求.
	 *
	 */
	class SampleConnector : public virtual simplemq::IMessageAdapter, public virtual RequestValidator,public std::enable_shared_from_this<SampleConnector>
	{
	private:
		/** 当前设备的MAC地址HEX(16进制)字符串 */
		std::string selfMac;
		/** 当前连接的CLIENT端MAC地址 */
		std::string connectedMAC;
		/** 频道取消注册时，将{@link #connectedMAC}复位 */
		void unregistedListener(const simplemq::Channel & ch)
		{
			connectedMAC = "";
		}
		simplemq::publisher_ptr ackPublisher;
		simplemq::subscriber_ptr subscriber;
		/** 当前连接的CLIENT端的请求频道 */
		std::string requestChannel;
		std::chrono::milliseconds idleTimeLimit;
		std::chrono::milliseconds timerPeriod;
		item_driver_ptr itemAdapter;
		req_validator_ptr requestValidator;
		CppTime::Timer timer;
		dev_info_ptr devInfo;
	public:
		SampleConnector(const simplemq::publisher_ptr & publisher, const simplemq::subscriber_ptr & subscriber)
			: ackPublisher(publisher), subscriber(subscriber),
			timerPeriod(std::chrono::milliseconds(2000)),
			idleTimeLimit(std::chrono::milliseconds(DEFAULT_IDLE_TIME_MILLS))
		{
			throw_if_null(publisher);
			throw_if_null(subscriber);
			timer.add(timerPeriod, [this](CppTime::timer_id) {
				try
				{
					auto c = this->subscriber->getChannel(this->requestChannel);
					if (c)
					{
						auto adapter = std::dynamic_pointer_cast<BaseItemDriver>(c->adapter);
						throw_if_null(adapter);
						auto lastHit = adapter->lastHitTime();
						if (lastHit.time_since_epoch().count() && (std::chrono::system_clock::now() - lastHit > idleTimeLimit))
						{
							this->subscriber->unregister({ requestChannel });
							this->requestChannel = "";
						}
					}
				}
				catch (const std::exception& e)
				{
					SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
				}
				catch (...)
				{
					SAMPLE_ERR("UNKNOW ERROR");
				}
			}, timerPeriod);
		}
		virtual ~SampleConnector() = default;
		/**
		 * 将连接请求字符串解析为{@link ConnectReq}对象，通过验证密码来确定连接是否有效
		 */
		virtual void validate(const simplemq::json &connj, std::string &clientID) override
		{
			ConnectReq req(connj);
			clientID = req.mac;
			throw_if_msg(req.mac.empty(), "NULL REQUEST MAC ADDRESS");
			throw_if_msg(req.pwd.empty(), "NULL REQUEST PASSWORD");
			if (devInfo)
			{
				auto admPwd = devInfo->getPassword();
				throw_if_msg(admPwd.empty(), "admin password for device is null");
				auto pwdmd5 = md5::digestString(admPwd);
				/** 比较password的MD5 */
				throw_except_if_msg(std::logic_error, gdface::tolower(req.pwd) != gdface::tolower(pwdmd5), "INVALID REQUEST PASSWORD");
			}
		}
		/**
		 * 处理来自管理端的连接请求<br>
		 * 如果收到的数据无法解析成{@link ConnectReq}实例则忽略
		 * 如果连接密码不匹配或其他管理端已经连接返回错误信息,否则返回随机生成的管理操作频道名。
		 * @param connj 连接请求
		 */
		virtual void onSubscribe(const simplemq::json &connj) override
		{
			Ack ack;
			ack.status = Status::ACK_OK;
			ack.deviceMac = selfMac;
			std::string ackChannel;
			/** 请求端的MAC地址或其他唯一识别ID */
			std::string clientIDRef;
			try
			{
				getRequestValidator()->validate(connj, clientIDRef);
				std::string reqMAC = clientIDRef;
				throw_except_if_msg(std::logic_error, reqMAC.empty(), "VALIDATE FAIL");
				ackChannel = getAckChannel(reqMAC);
				throw_except_if_msg(std::logic_error, !connectedMAC.empty() && connectedMAC != reqMAC, "ANOTHER CLIENT LOCKED");
				connectedMAC = reqMAC;
				/** 密码匹配则发送请求频道名 */
				if (requestChannel.empty()) 
				{
					/** 生成请求频道名 */
					auto now = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
					throw_except_if_msg(std::logic_error, nullptr == devInfo, "devInfo is null,please call setDeviceInfo to set");
					requestChannel = gdface::string_utils::format("%s_dtalk_%d", devInfo->getMacAsString().c_str(), int(now.time_since_epoch().count()& 0xffff));
				}
				/** 请求频道名作为响应消息返回值 */
				ack.value = requestChannel;
				if (!subscriber->getChannel(requestChannel))
				{
					throw_if_msg(!itemAdapter, "Dtalk ENGINE NOT READY");
					itemAdapter->setAckChannel(ackChannel);
					/** 必须在另开线程执行注册，否则会造成onSubscribe调用者JedisPubSub状态异常 */
					std::thread _run([ackChannel,this]() {
						try
						{
							/** 订阅请求频道用于命令发送 */
							simplemq::Channel c(requestChannel, itemAdapter);
							c.addUnregistedListener(std::bind(&SampleConnector::unregistedListener, this, std::placeholders::_1));
							subscriber->_register({ c });
							SAMPLE_LOG("Connect created(建立连接)for client:{}", connectedMAC);
							SAMPLE_LOG("request channel {} \nack channel       {} ", c.name, ackChannel);
						}
						catch (const std::exception& e)
						{
							SAMPLE_ERR("ERROR {}:{}", typeid(e).name(), e.what());
						}
						catch (...)
						{
							SAMPLE_ERR("UNKNOW ERROR");
						}
					});
					_run.join();
				}
			}
			catch (const nlohmann::detail::type_error &)
			{
				/** 忽略无法解析成ConnectReq请求对象的数据 */
				SAMPLE_LOG("REJECT REQUEST {}", connj.dump());
			}
			catch (const nlohmann::detail::exception &)
			{
				/** 忽略无法解析成ConnectReq请求对象的数据 */
				SAMPLE_LOG("REJECT REQUEST {}", connj.dump());
			}
			catch (const std::exception& e)
			{
				ack.status = Status::ACK_ERROR;
				ack.statusMessage = e.what();
			}
			catch (...)
			{
				ack.status = Status::ACK_ERROR;
				ack.statusMessage = "UNKNOW ERROR";
			}
			if (ackChannel.empty() && !clientIDRef.empty()) 
			{
				ackChannel = getAckChannel(clientIDRef);
			}
			if (!ackChannel.empty()) {
				/** 向响应频道发送响应消息 */
				simplemq::Channel channel(ackChannel);
				ackPublisher->publish(channel.name, ack);
			}
		}
		sample_onnector_ptr setItemAdapter(const item_driver_ptr& adapter) {
			this->itemAdapter = adapter;
			if (this->itemAdapter) {
				this->itemAdapter->setSelfMac(selfMac);
			}
			return shared_from_this();
		}
		/**
		* 设置空闲时间限制(毫秒),超过此时间,自动中断连接
		* @param idleTimeLimit 空闲时间限制(毫秒)
		* @return 当前对象
		*/
		sample_onnector_ptr setIdleTimeLimit(const std::chrono::milliseconds& idleTimeLimit) {
			if (idleTimeLimit.count() > 0) {
				this->idleTimeLimit = idleTimeLimit;
			}
			return shared_from_this();
		}
		/**
		 * 设置定义检查连接的任务时间间隔(毫秒)
		 * @param timerPeriod 时间间隔(毫秒)
		 * @return 当前对象
		 */
		sample_onnector_ptr setTimerPeriod(const std::chrono::milliseconds &timerPeriod) {
			if (timerPeriod.count() > 0) {
				this->timerPeriod = timerPeriod;
			}
			return shared_from_this();
		}
		req_validator_ptr getRequestValidator() {			
			return requestValidator ? requestValidator : shared_from_this();
		}

		/**
		 * 设置连接请求验证接口实例，如果不指定，默认使用基于{@link ConnectReq}格式的请求验证
		 * 参见 {@link #validate(String, AtomicReference)}
		 * @param requestValidator 不可为{@code null}
		 * @return 当前对象
		 */
		sample_onnector_ptr setRequestValidator(const req_validator_ptr &requestValidator) {
			throw_if_null(requestValidator);
			this->requestValidator = requestValidator;
			return shared_from_this();
		}
		/**
		 * @return 返回当前设备的MAC地址(HEX字符串)
		 */
		std::string getSelfMac() const{
			return selfMac;
		}
		/**
		 * 设置当前设备的MAC地址(HEX字符串)
		 * @param selfMac 要设置的 selfMac
		 * @return 当前对象
		 */
		sample_onnector_ptr setSelfMac(const std::string &selfMac)
		{
			this->selfMac = selfMac;
			return shared_from_this();
		}
		sample_onnector_ptr setDeviceInfo(const dev_info_ptr &devInfo)
		{
			this->devInfo = devInfo;
			return shared_from_this();
		}
	};
}/** namespace engine */} /** namespace dtalk */
#endif /* DTALK_DTALK_ENGINE_H_ */
